package com.haisen.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.haisen.core.utils.SpringBeanUtil;
import com.haisen.dataobject.dto.JobNoDTO;
import com.haisen.dataobject.dto.PubSubMsgDTO;
import com.haisen.dataobject.entity.JobNo;
import com.haisen.dataobject.entity.VerifyCodeHist;
import com.haisen.dataobject.entity.VerifyCodeTmp;
import com.haisen.dataobject.mapper.base.JobNoMapper;
import com.haisen.dataobject.mapper.base.VerifyCodeHistMapper;
import com.haisen.dataobject.mapper.base.VerifyCodeTmpMapper;
import com.haisen.dataobject.mapper.base.VerifyPubSubMapper;
import com.haisen.service.JobNoService;
import com.haisen.service.VerifyCodeTmpService;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;
import tk.mybatis.mapper.weekend.Weekend;
import tk.mybatis.mapper.weekend.WeekendCriteria;

import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class VerifyCodeTmpServiceImpl implements VerifyCodeTmpService {

      @Autowired
      private VerifyCodeTmpMapper verifyCodeTmpMapper;

      @Autowired
      private JobNoService jobNoService;

      @Autowired
      private VerifyPubSubMapper verifyPubSubMapper;
//批量插入

    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Autowired
     private VerifyCodeHistMapper verifyCodeHistMapper;

    private static WebSocketService webSocketService ;

   private  List<JobNoDTO> jobNoMapperList = new ArrayList<>();
   // private List<VerifyCodeHist> verifyCodeHists = new ArrayList<>();

    @Override
    public void delete(Long[] ids) {
        Weekend<VerifyCodeTmp> example = Weekend.of(VerifyCodeTmp.class);
        WeekendCriteria<VerifyCodeTmp,Object> criteria = example.weekendCriteria();
        criteria.andIn(VerifyCodeTmp::getId, Arrays.asList(ids));
         verifyCodeTmpMapper.deleteByExample(example);
    }

    @Override
    public void scanning() {
        //1、获取要发送队列,一次扫描5条
           int top =5;
          List<VerifyCodeTmp> verifyCodeTmpList = getQuenceMsg(top);
          if (verifyCodeTmpList.size() < 1) return;
          //log.info("推送队列={}",JSON.toJSONString(verifyCodeTmpList));

        //2、获取订阅主题和发送对象
         List<PubSubMsgDTO> pubSubMsgDTOList = new ArrayList<>();

        //扫描程序，重复连接，好资源，如果有数据才连
        //todo：是否写入到缓存中，下部优化处理
        //jobNoMapperList = jobNoMapper.selectAll();
        //采用缓存试下
          jobNoMapperList = JSON.parseArray(jobNoService.getAllJobNoINCache(),JobNoDTO.class);
         for(VerifyCodeTmp verifyCodeTmp : verifyCodeTmpList){
             //log.info("推送:{}",JSON.toJSONString(getPubSubDTO(verifyCodeTmp)));
             pubSubMsgDTOList.add(getPubSubDTO(verifyCodeTmp));
         }
        //遍历
         List<VerifyCodeHist> verifyCodeHistList = new ArrayList<>();
        for (PubSubMsgDTO pubSubMsgDTO : pubSubMsgDTOList) {
             sendToSub(pubSubMsgDTO);
              if (pubSubMsgDTO.getCtype().equals("all")){
                  verifyCodeHistList.add(new VerifyCodeHist("all",pubSubMsgDTO.getPublisherName(),pubSubMsgDTO.getCtype(),pubSubMsgDTO.getContent()));
              }else {
                  for(String u : pubSubMsgDTO.getSubscribers()){
                      verifyCodeHistList.add(new VerifyCodeHist(u,pubSubMsgDTO.getPublisherName(),pubSubMsgDTO.getCtype(),pubSubMsgDTO.getContent()));
                  }
              }
           }

       // log.info("已发送消息：{}",JSON.toJSONString(pubSubMsgDTOList));
       // log.info("保存信息={}",JSON.toJSONString(verifyCodeHistList));
         creatBatch(verifyCodeHistList);
         //删除已发送的记录
        Long[] verifyCodeTmpIds = verifyCodeTmpList.stream()
                .map(o -> {
                    return o.getId();
                }).collect(Collectors.toList()).toArray(new Long[verifyCodeTmpList.size()]);
        log.info("删除已发信息IDs={}",JSON.toJSONString(verifyCodeTmpIds));
         delete(verifyCodeTmpIds);

    }


    @Override
    public PubSubMsgDTO getPubSubDTO(VerifyCodeTmp verifyCodeTmp) {

        PubSubMsgDTO pubSubMsgDTO = new PubSubMsgDTO();
        //先将发布者赋值，不管是否在工号里面,默认群发方式
        pubSubMsgDTO.setPublisher(verifyCodeTmp.getPublisher());
        pubSubMsgDTO.setCtype("p2p");
        pubSubMsgDTO.setContent(verifyCodeTmp.getContent());
        // log.info("工号数量={}",jobNoMapperList.size());
         //发布者的Id
       List<Long> publishIdList = jobNoMapperList.stream().filter( o->{
          // log.info("jobNO boss工号={}，和商汇={},消息发布者={}",o.getAcccount4a(),o.getAccountNo(),verifyCodeTmp.getPublisher());
             if ((o.getAcccount4a() != null && o.getAcccount4a().equals(verifyCodeTmp.getPublisher()))|| (o.getAccountNo() != null && o.getAccountNo().equals(verifyCodeTmp.getPublisher()))) return true;
             else return false;
         }).map(o -> {
              pubSubMsgDTO.setPublisherName(o.getChineseNo());
             return o.getId();
         }).collect(Collectors.toList());

       //如果发布者不在工号列表中，非验证码，进行广播群发
         if( publishIdList.size() <1){
              pubSubMsgDTO.setCtype("all");
              pubSubMsgDTO.getSubscribers().add("all");
             return pubSubMsgDTO;
         }

       Long publisId =  publishIdList.get(0);
      log.info("发布者ID={}",publisId);
       //查找发布者和订阅者关系列表
        //todo：如果没有人订阅，如何处理,不用过滤，都要处理消息
        //1、如果有人订阅，点对点发
        List<String> subList = verifyPubSubMapper.selectAll().stream()
                .filter(o -> {
                    log.info("idList={},publishId={}",JSON.toJSONString(o.getPublishIdList()),publisId);
                    if (o.getPublishIdList().contains(publisId)) return true;
                    else return false;
                }).map(o -> {
                     //添加发送用户
                     pubSubMsgDTO.getSubscribers().add(o.getSubscribe());
                     log.info("定制用户:{}，点对点发送",o.getSubscribe());
                    return o.getSubscribe();
                }).collect(Collectors.toList());
        //2、如果所有人都没有订阅，群发，或发给指定人
        //dto getSubscribers()要赋值，如果为空，则为null
          if (subList.size()<1){
              pubSubMsgDTO.setCtype("all");
              pubSubMsgDTO.getSubscribers().add("all");
          }
     //   log.info("消息：{},推送= {}",pubSubMsgDTO.getContent(), JSON.toJSONString(pubSubMsgDTO.getSubscribers()));
       // log.info("已推送记录：{}",JSON.toJSONString(verifyCodeHists));
        return pubSubMsgDTO;
    }

    @Override
    public List<VerifyCodeTmp> getQuenceMsg(int top) {
       /*  Weekend<VerifyCodeTmp> example = Weekend.of(VerifyCodeTmp.class);
         example.setOrderByClause("id limit "+top);
         return verifyCodeTmpMapper.selectByExample(example);*/
       return verifyCodeTmpMapper.selectTop(top);
    }

    //发送消息
    @Override
    public void sendToSub(PubSubMsgDTO pubSubMsgDTO) {
        webSocketService = SpringBeanUtil.getBean(WebSocketService.class);
        if (pubSubMsgDTO.getCtype().equals("all")) {
           // log.info("群发："+pubSubMsgDTO.getContent());
            webSocketService.sendMessage(pubSubMsgDTO.getContent());
        } else {
            for(String subscriber : pubSubMsgDTO.getSubscribers()) {
               // log.info("p2p:"+subscriber+"=>"+pubSubMsgDTO.getContent());
                JSONObject jsonObject = new JSONObject();
                 jsonObject.put("publisher",pubSubMsgDTO.getPublisher());
                 jsonObject.put("publisherName",pubSubMsgDTO.getPublisherName());
                 String[] contentArray = pubSubMsgDTO.getContent().split(",");
                 jsonObject.put("time",contentArray[1]);
                 if (contentArray.length >3){
                     jsonObject.put("content",contentArray[4]);
                 }else {
                     jsonObject.put("content",pubSubMsgDTO.getContent());
                 }
                webSocketService.SendMessageToUser(subscriber, JSON.toJSONString(jsonObject));
            }
        }
    }

    public void creatBatch(List<VerifyCodeHist> list){
        SqlSession sqlSession= sqlSessionFactory.openSession(ExecutorType.BATCH);
        Integer lineNumber = 0;
        Integer insertedNumer = 0;
        int batchCount = 200;// 每批commit的个数
        int batchLastIndex = batchCount;// 每批最后一个的下标
        //设置保存的条数
        lineNumber = list.size();
        try {
            for (int index = 0; index < list.size(); ) {
                if (batchLastIndex >= list.size()) {
                    batchLastIndex = list.size();
                    list.subList(index, batchLastIndex).stream()
                            .map(o -> {
                                return verifyCodeHistMapper.insertSelective(o);
                            }).collect(Collectors.toList());
                    //设置已保存的数量
                    insertedNumer += batchLastIndex-index;
                    sqlSession.commit();
                    sqlSession.clearCache();
                    break;// 数据插入完毕，退出循环
                } else {
                    list.subList(index, batchLastIndex).stream()
                            .map(o -> {
                                return verifyCodeHistMapper.insertSelective(o);
                            }).collect(Collectors.toList());
                    //设置已保存的数量
                    insertedNumer += batchLastIndex-index;

                    index = batchLastIndex;// 设置下一批下标
                    sqlSession.commit();
                    sqlSession.clearCache();
                }
                //导入进度
                batchLastIndex = index + (batchCount - 1);
            }
        } finally {
            sqlSession.close();
        }

    }
}
